package com.atguigu.chapter06.windowtomysql;

import com.atguigu.chapter05.pojo.Event;
import com.atguigu.chapter05.source.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.util.Calendar;
import java.util.HashMap;
import java.util.Map;


public class Window_MySQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.socketTextStream("whost", 22422)
                .map(new MapFunction<String, Event>() {
                    @Override
                    public Event map(String value) throws Exception {
                        String[] strs = value.split(" ");
                        return new Event(strs[0], strs[1], Calendar.getInstance().getTimeInMillis());
                    }
                })
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event event, long l) {
                                return event.timestamp;
                            }
                        }))
                .keyBy(event -> event.user)
                .window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1)))
                .aggregate(new AggregateFunction<Event, Map<String, Integer>, Map<String, Integer>>() {
                    @Override
                    public Map<String, Integer> createAccumulator() {
                        return new HashMap<>();
                    }

                    @Override
                    public Map<String, Integer> add(Event event, Map<String, Integer> stringSetMap) {
                        // 如果存在则取map.get(event.user)，否则new一个新的Set
                        String user = event.user;
                        stringSetMap.put(user, stringSetMap.getOrDefault(user, 0) + 1);
                        return stringSetMap;
                    }

                    @Override
                    public Map<String, Integer> getResult(Map<String, Integer> stringSetMap) {
                        return stringSetMap;
                    }

                    @Override
                    public Map<String, Integer> merge(Map<String, Integer> stringSetMap, Map<String, Integer> acc1) {
                        return null;
                    }
                })
                .map(new MapFunction<Map<String, Integer>, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(Map<String, Integer> stringIntegerMap) throws Exception {
                        int maxCount = Integer.MIN_VALUE;
                        Tuple2<String, Integer> mapResult = new Tuple2<>();
                        for (Map.Entry<String, Integer> entry : stringIntegerMap.entrySet()) {
                            if (entry.getValue() > maxCount) {
                                maxCount = entry.getValue();
                                mapResult.f0 = entry.getKey();
                                mapResult.f1 = entry.getValue();
                            }
                        }
                        return mapResult;
                    }
                })
                .addSink(
                        JdbcSink.sink("INSERT INTO window_data (name, count) VALUES (?, ?)",
                                (statement, tuple) -> {
                                    statement.setString(1, tuple.f0);
                                    statement.setInt(2, tuple.f1);
                                },
                                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                        .withUrl("jdbc:mysql://whost:3306/flinkdemo?useUnicode=true&characterEncoding=utf-8&serverTimezone=UTC")
                                        .withDriverName("com.mysql.cj.jdbc.Driver")
                                        .withUsername("root")
                                        .withPassword("cao123")
                                        .build()
                        ));
        env.execute();
    }
}
